package com.overlord.im.commons.mq.config;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.Message;

/**
 * @ClassName ManualAckInterceptor
 * @Author wuwei
 * @Description NONE
 * @Date 2020/4/18 11:01
 **/
@Slf4j
public class ManualAckInterceptor implements MethodInterceptor {

    public Object invoke(MethodInvocation invocation) throws Throwable {
        Object proceed = null;
        try {
            proceed = invocation.proceed();
        } catch (Throwable throwable) {
            if (throwable.getCause() instanceof ImmediateAcknowledgeAmqpException) {
                Object[] arguments = invocation.getArguments();
                Channel channel = (Channel) arguments[0];
                Message message = (Message) arguments[1];
                ack(message, channel);
            }
            throw throwable;
        }
        return proceed;
    }

    /**
     * 手动ack
     *
     * @param message
     * @param channel
     */
    private void ack(Message message, Channel channel) {
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("重试失败后手动ack消息:{}", message.getMessageProperties().getMessageId());
        } catch (Exception e) {
            log.error("ack消息异常", e);
        }
    }
}
